package com.example.mq.mqserver.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

// 通过这个类, 来实现消费消息的核心逻辑
public class ConsumerManager {
    // 持有上层的 VirtualHost 对象的引用. 用来操作数据.
    private VirtualHost parent;
    // 指定一个线程池, 负责去执行具体的回调任务
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);
    // 存放令牌的队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    // 扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost p) {
        parent = p;

        // 启动扫描线程
        scannerThread = new Thread(() -> {
            while(true) {
                try {
                    // 1.拿到令牌
                    String queueName = tokenQueue.take();
                    // 2.根据令牌, 找到队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if(queue == null) {
                        throw new MqException(("[ConsumerManager - ConsumerManager] 构造方法出错, 扫描线程启动中, 令牌对应队列查询不到! queueName=" + queueName));
                    }
                    // 3.从这个队列中消费一个消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
        scannerThread.setDaemon(true); // 设置为后台线程
        scannerThread.start();
    }

    /**
     * 通知消费者消费消息
     * 该方法的调用时机 就是 发送消息的时候
     * 消息发送完毕后, 把通知要发送的队列, 往阻塞队列里插入, 扫描线程会扫描阻塞队列, 寻找要需要消费的队列, 并进行消费
     * @param queueName
     * @throws InterruptedException
     */
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }


    /**
     * 消费者订阅某个队列的消息
     * @param consumerTag 消费者标识
     * @param queueName 队列名
     * @param autoAck 消费完成后, 回应方式
     * @param consumer 消费者消费消息的方式, 具体内容
     */
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        // 找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if(queue == null) {
            throw new MqException(("[ConsumerManager - addConsumer] 消费者订阅某个队列中的消息失败, 订阅的队列不存在! queueName=" + queueName));
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);
            // 如果当前队列中已经有了一些消息, 则需要立即消费掉
            int n = parent.getMemoryDataCenter().getMassageCount(queueName);
            for(int i=0;i<n;i++) {
                consumeMessage(queue);
            }
        }
    }

    /**
     * 消费者一个消费消息
     * @param queue
     */
    private void consumeMessage(MSGQueue queue) {
        // 1.先按照轮询的方式, 找到一个消费者
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if(luckyDog == null) {
            // 如果队列没有消费者, 那就不消费消息.
            return ;
        }
        // 2.从队列中取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message == null) {
            // 当前队列中没有消息, 也不用消费消息
            return ;
        }
        // 3.把消息带入到消费者的回溯方法中, 丢给线程池执行
        workerPool.submit(() -> {
            try {
                // 1.把消息放到待确认的集合中, 这个操作势必在执行回调之前
                parent.getMemoryDataCenter().addMessageWithAck(queue.getName(), message);

                // 2.真正执行回调操作
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(),
                        message.getBody());

                // 3.如果当前是 "自动应答", 就可以直接把消息删除
                //   如果当前是 "手动应答", 则先不处理, 交给后续消费者调用 basicAck 方法来处理
                if(luckyDog.isAutoAck()) {
                    // 1)删除硬盘上的消息
                    if(message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    // 2)删除上面的待确认集合中的消息
                    parent.getMemoryDataCenter().removeMessageWithAck(queue.getName(), message.getMessageId());
                    // 3)删除内存中消息中心里的消息
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println("[ConsumerManager - consumeMessage] 消费者消费消息成功! queueName=" + queue.getName());
                }
            } catch (Exception e) {
                System.out.println("[ConsumerManager - consumeMessage] 消费者消费消息失败! queueName=" + queue.getName());
                e.printStackTrace();
            }
        });
    }
}
